package com.demo.rocketmq.sender;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

import java.util.List;


/**
 * 发送顺序消息
 */
@Slf4j
@Configuration
public class SequenceMqSender {

    @Autowired
    private DefaultMQProducer defaultMQProducer;

    /**
     * 发送消息
     * keys业务唯一标识作为幂等处理
     * 参考：https://help.aliyun.com/document_detail/44397.htm?spm=a2c4g.11186623.2.13.618b7907ZyouGT#concept-2047147
     *
     * @param msg
     * @param tags
     * @param keys
     */
    public SendResult sendMsg(String topic, String msg, String tags, String keys, Long id) throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
        log.info("MQ发送顺序消息，topic = {}，tags = {}，keys = {}，msg = {}", topic, tags, keys, msg);
        Message message = new Message(topic, tags, keys, msg.getBytes());
        return defaultMQProducer.send(message, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                Long id = (Long) arg;
                long index = id % mqs.size();
                return mqs.get((int) index);
            }
        }, id);
    }


}
